package intensivecomp.util;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;
import org.apache.hadoop.util.LineReader;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class CustomerNLineInputFormat extends FileInputFormat<Text, Text> {
    public static final String LINES_PER_MAP = "mapreduce.input.lineinputformat.linespermap";

    public CustomerNLineInputFormat() {
    }

    public RecordReader<Text, Text> createRecordReader(InputSplit genericSplit, TaskAttemptContext context) throws IOException {
        context.setStatus(genericSplit.toString());
        return new WholeSplitRecordReader();
    }

    public List<InputSplit> getSplits(JobContext job) throws IOException {
        List<InputSplit> splits = new ArrayList();
        int numLinesPerSplit = getNumLinesPerSplit(job);
        Iterator var4 = this.listStatus(job).iterator();

        while(var4.hasNext()) {
            FileStatus status = (FileStatus)var4.next();
            splits.addAll(getSplitsForFile(status, job.getConfiguration(), numLinesPerSplit));
        }

        return splits;
    }

    public static List<CustomerFileSplit> getSplitsForFile(FileStatus status, Configuration conf, int numLinesPerSplit) throws IOException {
        List<CustomerFileSplit> splits = new ArrayList();
        Path fileName = status.getPath();
        if (status.isDirectory()) {
            throw new IOException("Not a file: " + fileName);
        } else {
            FileSystem fs = fileName.getFileSystem(conf);
            LineReader lr = null;

            try {
                FSDataInputStream in = fs.open(fileName);
                lr = new LineReader(in, conf);
                Text line = new Text();
                int numLines = 0;
                long begin = 0L;
                long length = 0L;
                boolean var14 = true;

                //每次取两行，形式为：1,2；2,3,；3,4...
                int num;
                int skipLinesNum = 6;
                for (int i =0;i<skipLinesNum; i++){
                    num = lr.readLine(line);
                    begin += num;
                }
                while((num = lr.readLine(line)) > 0) {
                    ++numLines;
                    length += (long)num;
                    if (numLines == numLinesPerSplit) {
                        splits.add(createFileSplit(fileName, begin, length));
                        System.out.println(splits.get(splits.size()-1));
                        length -= (long)num;
                        begin += length;
                        length = 0L;
                        length += (long)num;
                        numLines = 1;
                        num = 0;
                    }
                }

                if (num > 0) {
                    splits.add(createFileSplit(fileName, begin, length));
                }
            } finally {
                if (lr != null) {
                    lr.close();
                }

            }

            return splits;
        }
    }

    protected static CustomerFileSplit createFileSplit(Path fileName, long begin, long length) {

        CustomerFileSplit split ;
        if (begin == 0L)
            split = new CustomerFileSplit(fileName, begin, length - 1L, new String[0]) ;
        else
            split = new CustomerFileSplit(fileName, begin - 1L, length, new String[0]);
        return  split;
//        return begin == 0L ? new FileSplit(fileName, begin, length - 1L, new String[0]) : new FileSplit(fileName, begin - 1L, length, new String[0]);
    }

    public static void setNumLinesPerSplit(Job job, int numLines) {
        job.getConfiguration().setInt("mapreduce.input.lineinputformat.linespermap", numLines);
    }

    public static int getNumLinesPerSplit(JobContext job) {
        return job.getConfiguration().getInt("mapreduce.input.lineinputformat.linespermap", 1);
    }
}
